package com.zzz.java8;

import cn.hutool.core.thread.ThreadFactoryBuilder;
import org.junit.Test;

import java.util.concurrent.*;

/**
 * @author zhuzhizun
 * @date 2021/9/9
 */
public class CompletableFutureTest {


    @Test
    public void test01() throws ExecutionException, InterruptedException {
        // runAsync 执行Runnable接口  不关心返回值
        CompletableFuture.runAsync(() -> {
            System.out.println("Hello CompletableFuture");
            ;
        });

        //supplyAsync 执行供给型函数接口  有返回值
        System.out.println(CompletableFuture.supplyAsync(() -> "Hello").get());
    }

    @Test
    public void test02() {
        //thenApply 供给型接口参数 默认使用上一个的线程池 也可以自定义线程池
        CompletableFuture.supplyAsync(() -> "Hello").thenApply(s -> s + "World");

        //thenApplyAsync 供给型接口参数 默认使用ForkJoinPool
        CompletableFuture.supplyAsync(() -> "Hello").thenApplyAsync(s -> s + "World");

        //thenAcceptAsync 消费型接口参数 可以获取到上一步的结果
        CompletableFuture.completedFuture("Hello").thenApply(s -> s + " zzz").thenAcceptAsync(System.out::println);

        //thenRun Runnable参数 获取不到上一步的结果
        CompletableFuture.completedFuture("Hello").thenApply(s -> s + " zzz").thenRun(() -> System.out.println("zzz")
        );

    }

    @Test
    public void test03() {
        // whenComplete BiConsumer消费接口 提供两个消费 r表示结果 e表示异常
        CompletableFuture.completedFuture("Hello").whenComplete((r, e) -> {
            System.out.println(r);
            System.out.println(e);
        });

        CompletableFuture.supplyAsync(() -> {
            if (true) {
                throw new RuntimeException("Completable exception");
            }
            return "Hello";
        }).whenComplete((r, e) -> {
            System.out.println(r);
            System.out.println(e);
        });

        CompletableFuture.supplyAsync(() -> {
            if (true) {
                throw new RuntimeException("Completable exception");
            }
            return "Hello";
        }).handle((r, e) -> {
            System.out.println(e);
            return r != null ? r : "world!";
        }).thenAccept(System.out::println);
    }

    @Test
    public void test04() throws InterruptedException, ExecutionException, TimeoutException {
        ThreadFactory threadFactory = ThreadFactoryBuilder
                .create().setNamePrefix("zzz-thread-").build();
        ThreadPoolExecutor myThreadPool = new ThreadPoolExecutor(2,
                10,
                60,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(20),
                threadFactory,
                new ThreadPoolExecutor.DiscardPolicy()
        );


        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            return "Hello";
        }, myThreadPool);

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            return "Hello2";
        }, myThreadPool);

        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            return "Hello3";
        }, myThreadPool);

        CompletableFuture<String> future4 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            return "Hello4";
        }, myThreadPool);
        System.out.println(future.get(1,TimeUnit.SECONDS));
        System.out.println(future2.get(1,TimeUnit.SECONDS));
        System.out.println(future3.get(1,TimeUnit.SECONDS));
        System.out.println(future4.get(1,TimeUnit.SECONDS));









        CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            return "Hello";
        }, myThreadPool).thenApply((s) -> {
            System.out.println(Thread.currentThread().getName());
            return s + " ZZZ";
        }).thenComposeAsync((s) -> {  //thenCompose 可以结合两个Completable 前一个结果作为参数传到后面一个
            System.out.println(Thread.currentThread().getName());
            return CompletableFuture.supplyAsync(() -> s + " World!");
        }).thenAccept(System.out::println);
        myThreadPool.shutdown();

        //thenCombine 将两个Completable结果合并 返回一个新的Completable 两者之间没有先后顺序
        CompletableFuture.supplyAsync( () -> "Hello")
                .thenCombine(CompletableFuture.supplyAsync(() -> " World"),(s1,s2) -> s1+s2)
                .thenAccept(System.out::println);
    }

    @Test
    public void test05(){
        CompletableFuture<Integer> task1 = CompletableFuture.supplyAsync(() -> {
            System.out.println("task1........");
            return 1;
        });
        CompletableFuture<Integer> task2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("task2........");
            return 2;
        });
        CompletableFuture<Integer> task3 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("task3........");
            return 3;
        });
        CompletableFuture<Integer> task4 = CompletableFuture.supplyAsync(() -> {
            System.out.println("task4........");

            return 4;
        });

/*        CompletableFuture<Void> allOf = CompletableFuture.allOf(task1, task2, task3, task4);
        //allOf 所有的任务都完成了才返回
        allOf.join();
        System.out.println("allOf down.......");*/


        CompletableFuture<Object> anyOf = CompletableFuture.anyOf(task1, task2, task3, task4);
        //anyOf 不必等到所有的任务都完成才返回
        anyOf.join();
        System.out.println("anyOf down.......");

    }


}
